import java.util.Properties

import org.apache.kafka.clients.consumer
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Q1_SparkStreaming {


  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(2))
    val sc = ssc.sparkContext
    sc.setLogLevel("WARN")

    val groupID = "mygroup01"
    val dstream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Array("topicB"),
          getKafkaParams(groupID)))

    dstream
      .map(record => record.key().split(",").map(_.stripPrefix("<<<!>>>").stripSuffix("<<<!>>>")).reduce(_ + "|" + _))
      .foreachRDD { rdd => {
        //println(s"${rdd.count} --- ${time}")
        rdd.foreachPartition { s_iter =>
          val prop = new Properties()
          val brokers = "192.168.80.131:9092,192.168.80.132:9092,192.168.80.133:9092"
          val topic = "topicC"
          prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
          prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
          prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
          val producer = new KafkaProducer[String, String](prop)
          s_iter.foreach(s => producer.send(new ProducerRecord[String, String](topic, s, s)))
          producer.close()
          }
      }
      }
    ssc.start()
    ssc.awaitTermination()
  }

  def getKafkaParams(groupID: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "192.168.80.131:9092,192.168.80.132:9092,192.168.80.133:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupID,
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest",
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
    )
  }
}
